package cn.pink.gatesrv;

import cn.pink.common.constant.GlobalConst;
import cn.pink.common.proto.Msg;
import cn.pink.common.proto.MsgIds;
import cn.pink.common.support.tools.cache.CacheTool;
import cn.pink.common.support.tools.msg.MsgTool;
import cn.pink.common.support.utils.TimeUtils;
import cn.pink.core.*;
import cn.pink.core.config.CacheConfig;
import cn.pink.core.config.IronConfig;
import cn.pink.core.config.NetConfig;
import cn.pink.core.gen.proxy.DistrClass;
import cn.pink.core.gen.proxy.DistrMethod;
import cn.pink.core.support.*;
import cn.pink.gamesrv.RPCProxy;
import io.netty.buffer.ByteBuf;
import io.netty.channel.Channel;
import io.netty.handler.codec.http.websocketx.BinaryWebSocketFrame;

import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.util.concurrent.LinkedBlockingQueue;

//验证session逻辑
//取session的状态 0未登录:登录逻辑, 1游戏中:直接转发
//玩家在线一台设备, 又使用其他设备重复登录
//滚服模式:
//集群或分布式模式:
/**
 * 相当于一个长连接的包装, 放到port中统一管理
 * 1个session对应一个sessionService
 * @Author: pink
 * @Date: 2022/7/1 15:48
 */
@DistrClass(singleton = false, importClass = {Chunk.class, CallPoint.class})
public class SessionService extends Service {
    /** session */
    private Session session;
    /** 连接Channel */
    private final Channel channel;

    /** 玩家账号 */
    private String account;
    /** humanId */
    private String humanId;

    /** 清理中 */
    private boolean clearing = false;

    //连接验证
    /** 连接检查计时器 5秒检查一次 */
    private final TickTimer connCheckTickTimer = new TickTimer(5 * TimeUtils.SEC);
    /** 累计连续未收到返回的次数 */
    private int connCheckIncreaseTimes = 0;
    //end

    /** 待处理数据 */
    private final LinkedBlockingQueue<byte[]> input = new LinkedBlockingQueue<>();

    /** 延迟清理定时器 */
    private final TickTimer timerDelayClear = new TickTimer();
    /** 断开连接时是否立即清理数据 */
    private boolean closeImmediate = false;

    /** 上次收到客户端消息的时间 */
    private long lastMsgTime;

    /** 消息包计数器 */
    private int msgCounter = 0;
    /** 消息包开始计数的时间 */
    private long msgCountStartTime = 0;

    /** 保证消息接受不至于太夸张，如果1分钟内收到超过300个消息包，打印一条警告(可能遭受客户端攻击) */
    private static final int MSG_COUNT_MAX = 300;
    private static final long MSG_COUNT_TIME = TimeUtils.MIN;

    public SessionService(Port port, Session session, Channel channel) {
        super(port);
        this.session = session;
        this.channel = channel;

        this.lastMsgTime = port.getTimeCurrent();
    }

    @Override
    public Object getId() {
        return session.getId();
    }

    @Override
    public void pulseOverride() {
        long now = port.getTimeCurrent();

        // 统计，检测是否短期内收到大量的包(客户端攻击)
        pulseStatistics(now);

        //处理接收数据
        pulseInput();

        //连接验证
        connCheck();

        // 延时清理已到 开始清理
        if (timerDelayClear.isOnce(now)) {
            clear();
        }

        // 如果长时间没有收到客户端消息则关闭
        if ((lastMsgTime + IronConfig.GATE_PING_TIMEOUT < now) && channel.isOpen()) {
            channel.close();
            Log.gate.warn("long time no ping, close conn: id={}", session.getId());
        }
    }

    /**
     * 处理接收数据
     */
    private void pulseInput() {
        while (!input.isEmpty()) {
            inputHandler(input.poll());
        }
    }

    /**
     * 消息处理
     */
    private void inputHandler(byte[] msgBuf) {
        int msgId = Utils.bytesToBigEndian32(msgBuf, 4);

        //过滤不存在的协议
        if (MsgIds.getClassById(msgId) == null) {
            return;
        }

        lastMsgTime = port.getTimeCurrent();

        //处理ping
        if(msgId == MsgIds.CSPING) {
            sendMsg(MsgIds.SCPING, new Chunk(Msg.SCPing.newBuilder().setTime(lastMsgTime)));
            return;
        }

        Chunk msgChunk = new Chunk(msgBuf, 8, msgBuf.length - 8);

        //打印 message log
        MsgTool.getInstance().printMsg(msgId, msgChunk);

        //TODO 统计

        switch (session.getStatus()) {
            // 登录阶段
            case GlobalConst.SESSION_STATUS_LOGGING: {
                RPCProxy.AccountServiceProxy proxy = RPCProxy.AccountServiceProxy.newInstance();
                proxy.defaultHandler(session.getId(), msgBuf);
            }
            break;

            //游戏中
            case GlobalConst.SESSION_STATUS_PLAYING: {
                RPCProxy.HumanObjectServiceProxy proxy = RPCProxy.HumanObjectServiceProxy.newInstance(session.getGamePoint());
                proxy.defaultHandler(session.getId(), msgBuf);
            }
            break;

            //关闭
            case GlobalConst.SESSION_STATUS_CLOSED: {
                Log.gate.debug("接到客户端信息，但是玩家已是断线状态，忽略此消息。");
            }
            break;

            default:break;
        }
    }

    /**
     * 连接验证
     * 验证那些无用连接, 比如连上了啥也没干的.
     */
    private void connCheck() {
        if(IronConfig.GATE_CHECK_COUNT == 0) {
            return;
        }

        //验证间隔
        if (!connCheckTickTimer.isPeriod(Port.getTime())) {
            return;
        }

        //避免由于Debug断点等情况 造成瞬间发送多个检查请求
        connCheckTickTimer.reStart();

        //清理掉超时的连接
        if (connCheckIncreaseTimes >= IronConfig.GATE_CHECK_COUNT) {
            // 还处于连接状态，服务器强制断开连接
            if (channel.isOpen()) {
                channel.close();
            }

            //日志
            Log.gate.warn("clear error conn: conn={}, status={}, account={}, humanId={}", getId(), session.getStatus(), account, humanId);
            return;
        }

        //TODO 根据状态进行验证
        switch (session.getStatus()) {
            case GlobalConst.SESSION_STATUS_LOGGING:
                break;
            case GlobalConst.SESSION_STATUS_PLAYING:
                break;
            default: break;
        }

        //累加连接检查次数
        connCheckIncreaseTimes++;
    }

    /**
     * 统计，检测是否短期内收到大量的包(客户端攻击)
     */
    private void pulseStatistics(long now) {
        // 开始计时
        if (msgCounter == 0) {
            msgCountStartTime = now;
        }

        // 计数
        msgCounter += input.size();

        if (msgCounter > MSG_COUNT_MAX) {
            if (now - msgCountStartTime < MSG_COUNT_TIME) {
                Log.gate.warn("客户端发送消息过于频繁，{}毫秒内收到{}个消息包，ip={}", now - msgCountStartTime, msgCounter, getClientIp());
            }

            msgCounter = 0;
        }
    }

    /**
     * 获取客户端ip地址
     */
    private String getClientIp() {
        if (channel.remoteAddress() == null) {
            return "";
        }

        InetSocketAddress socketAddress = (InetSocketAddress) channel.remoteAddress();
        InetAddress ia = socketAddress.getAddress();
        return ia.getHostAddress();
    }

    /**
     * 立即清理玩家数据
     */
    private void clear() {
        if (clearing) {
            return;
        }

        // 设置为清理状态
        clearing = true;

        // 日志
        Log.gate.info("clear conn: connId={}, status={}", session.getId(), session.getStatus());

        // 根据登陆阶段 通知游戏服务器进行处理
        switch (session.getStatus()) {
            // 登陆阶段
            case GlobalConst.SESSION_STATUS_LOGGING:
                if(!IronConfig.GAME_LOGIN_ROLE) {
                    return;
                }

                //TODO 带角色登录逻辑

                break;

            //游戏阶段 根据session找到gamePoint 清掉内存
            case GlobalConst.SESSION_STATUS_PLAYING:
                RPCProxy.HumanObjectServiceProxy proxy = RPCProxy.HumanObjectServiceProxy.newInstance(session.getGamePoint());
                proxy.connClosed(session.getId());

                break;

            //关闭阶段
            case GlobalConst.SESSION_STATUS_CLOSED:
                break;

            default: break;
        }

        //清理sessionService
        port.delServiceBySafe(getId());
    }

    /**
     * channel收到数据
     */
    public void onChannelRead(byte[] data) {
        try {
            input.put(data);
        } catch (InterruptedException e) {
            throw new SysException(e);
        }
    }

    /**
     * channel连接断开
     */
    public void onChannelClosed() {
        // 正在清理中
        if (clearing || timerDelayClear.isStarted()) {
            return;
        }

        // 延迟关闭(等待重连)
        if (session.getStatus() == 1 && !closeImmediate) {
            timerDelayClear.start(IronConfig.GATE_RECONNECT_TIME);

            //TODO 通知HumanObject玩家已掉线，等待重连中
        } else {
            // 立即关闭
            clear();
        }
    }

    /**
     * 绑定human
     */
    private void bindHuman(String humanId) {
        this.humanId = humanId;
        CacheTool.getInstance().getCache(CacheConfig.SESSION).put(humanId, session);
        Log.gate.info("human login success: humanId={}, sessionPoint={}, gamePoint={}", humanId, session.getSessionPoint().toString(), session.getGamePoint().toString());
    }

    /**
     * 角色登陆回调 绑定session 通知客户端
     * 采用缓存-监听实现, 一个角色对应一个session, 如果监听到存在直接踢旧连接.
     */
    @DistrMethod
    public Param loginResult(String humanId, CallPoint humanPoint, Chunk msg) {
        session.setStatus(GlobalConst.SESSION_STATUS_PLAYING);
        session.setSessionPoint(new CallPoint(getPort().getNodeId(), getPort().getId(), getId()));
        session.setGamePoint(humanPoint);

        bindHuman(humanId);

        sendMsg(MsgIds.SCROLELOGIN, msg);

        return null;
    }

    /**
     * 发送消息至客户端
     */
    @DistrMethod
    public Param sendMsg(int msgId, Chunk msgBuf) {
        if (!channel.isActive()) {
            return null;
        }

        if (!channel.isWritable()) {
            return null;
        }

        // 调试用
        MsgTool.getInstance().printMsg(msgId, msgBuf);

        // TODO 信息统计

        int totalLength = msgBuf.length + 8;

        if(IronConfig.COMMON_NET.equals(NetConfig.WEBSOCKET.toString())) {
            ByteBuf buf = channel.alloc().buffer(totalLength);
            buf.writeIntLE(totalLength);
            buf.writeIntLE(msgId);
            buf.writeBytes(msgBuf.buffer);

            channel.writeAndFlush(new BinaryWebSocketFrame(buf));
        }
        else if(IronConfig.COMMON_NET.equals(NetConfig.SOCKET.toString())) {
            ByteBuf head = channel.alloc().buffer(8);
            head.writeInt(totalLength);
            head.writeInt(msgId);

            //写入头数据
            channel.write(head);
            //Chunk类型的msgbuf肯定是protobuf直接生成的 所以buffer属性中不会有多余数据 才能这么用
            //其余地方Chunk类不建议直接使用内部的buffer
            channel.write(msgBuf.buffer);
            channel.flush();
        }

        return null;
    }

    /**
     * 关闭连接
     */
    @DistrMethod
    public Param closeImmediate() {
        Log.gate.info("closeImmediate, sessionId={}", getId());

        port.addTask(new PortTask() {
            @Override
            public void execute(Port port) {
                // 立即关闭客户端
                closeImmediate = true;
                // 还处于连接状态，服务器强制断开连接
                if (channel.isOpen()) {
                    channel.close();
                } else {
                    clear();
                }
            }
        });

        return null;
    }
}
